package cn.dmp.tools

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 将bz2日志文件格式转换成parquet
  */
object Biz2Parquet {


    def main(args: Array[String]): Unit = {

        if(args.length !=2) {
            println(
                """
                  |cn.dmp.tools.Biz2Parquet
                  |参数不合法，请传递正确的参数个数：
                  |dataInputPath
                  |resultOutputPath
                """.stripMargin)
            sys.exit()
        }


        val sparkConf = new SparkConf()
          .setMaster("local[*]")
          .setAppName("日志转Parquet文件格式")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(sparkConf)

        // 接受参数
        val Array(dataInputPath, resultOutputPath) = args

        // 读取原始日志文件
        val rawData: RDD[String] = sc.textFile(dataInputPath)
        // 过滤非法数据
        val filtered: RDD[Array[String]] = rawData.map(_.split(",", -1)).filter(_.length >= 85)
        // Array[String] -> Row


        val data: RDD[Row] = filtered.map(arr => {

            import cn.dmp.beans.RichString._

            Row(
                arr(0),
                arr(1).toIntPlus,// String -> toIntPlus => RichString -> toIntPlus
                arr(2).toIntPlus,
                arr(3).toIntPlus,
                arr(4).toIntPlus,
                arr(5),
                arr(6),
                arr(7).toIntPlus,
                arr(8).toIntPlus,
                arr(9).toDoublePlus,
                arr(10).toDoublePlus,
                arr(11),
                arr(12),
                arr(13),
                arr(14),
                arr(15),
                arr(16),
                arr(17).toIntPlus,
                arr(18),
                arr(19),
                arr(20).toIntPlus,
                arr(21).toIntPlus,
                arr(22),
                arr(23),
                arr(24),
                arr(25),
                arr(26).toIntPlus,
                arr(27),
                arr(28).toIntPlus,
                arr(29),
                arr(30).toIntPlus,
                arr(31).toIntPlus,
                arr(32).toIntPlus,
                arr(33),
                arr(34).toIntPlus,
                arr(35).toIntPlus,
                arr(36).toIntPlus,
                arr(37),
                arr(38).toIntPlus,
                arr(39).toIntPlus,
                arr(40).toDoublePlus,
                arr(41).toDoublePlus,
                arr(42).toIntPlus,
                arr(43),
                arr(44).toDoublePlus,
                arr(45).toDoublePlus,
                arr(46),
                arr(47),
                arr(48),
                arr(49),
                arr(50),
                arr(51),
                arr(52),
                arr(53),
                arr(54),
                arr(55),
                arr(56),
                arr(57).toIntPlus,
                arr(58).toDoublePlus,
                arr(59).toIntPlus,
                arr(60).toIntPlus,
                arr(61),
                arr(62),
                arr(63),
                arr(64),
                arr(65),
                arr(66),
                arr(67),
                arr(68),
                arr(69),
                arr(70),
                arr(71),
                arr(72),
                arr(73).toIntPlus,
                arr(74).toDoublePlus,
                arr(75).toDoublePlus,
                arr(76).toDoublePlus,
                arr(77).toDoublePlus,
                arr(78).toDoublePlus,
                arr(79),
                arr(80),
                arr(81),
                arr(82),
                arr(83),
                arr(84).toIntPlus
            )
        })

        // 数据描述信息，表头
        val schema = StructType(
            Seq(
                StructField("sessionid",StringType),
                StructField("advertisersid",IntegerType),
                StructField("adorderid",IntegerType),
                StructField("adcreativeid",IntegerType),
                StructField("adplatformproviderid",IntegerType),
                StructField("sdkversion",StringType),
                StructField("adplatformkey",StringType),
                StructField("putinmodeltype",IntegerType),
                StructField("requestmode",IntegerType),
                StructField("adprice",DoubleType),
                StructField("adppprice",DoubleType),
                StructField("requestdate",StringType),
                StructField("ip",StringType),
                StructField("appid",StringType),
                StructField("appname",StringType),
                StructField("uuid",StringType),
                StructField("device",StringType),
                StructField("client",IntegerType),
                StructField("osversion",StringType),
                StructField("density",StringType),
                StructField("pw",IntegerType),
                StructField("ph",IntegerType),
                StructField("long",StringType),
                StructField("lat",StringType),
                StructField("provincename",StringType),
                StructField("cityname",StringType),
                StructField("ispid",IntegerType),
                StructField("ispname",StringType),
                StructField("networkmannerid",IntegerType),
                StructField("networkmannername",StringType),
                StructField("iseffective",IntegerType),
                StructField("isbilling",IntegerType),
                StructField("adspacetype",IntegerType),
                StructField("adspacetypename",StringType),
                StructField("devicetype",IntegerType),
                StructField("processnode",IntegerType),
                StructField("apptype",IntegerType),
                StructField("district",StringType),
                StructField("paymode",IntegerType),
                StructField("isbid",IntegerType),
                StructField("bidprice",DoubleType),
                StructField("winprice",DoubleType),
                StructField("iswin",IntegerType),
                StructField("cur",StringType),
                StructField("rate",DoubleType),
                StructField("cnywinprice",DoubleType),
                StructField("imei",StringType),
                StructField("mac",StringType),
                StructField("idfa",StringType),
                StructField("openudid",StringType),
                StructField("androidid",StringType),
                StructField("rtbprovince",StringType),
                StructField("rtbcity",StringType),
                StructField("rtbdistrict",StringType),
                StructField("rtbstreet",StringType),
                StructField("storeurl",StringType),
                StructField("realip",StringType),
                StructField("isqualityapp",IntegerType),
                StructField("bidfloor",DoubleType),
                StructField("aw",IntegerType),
                StructField("ah",IntegerType),
                StructField("imeimd5",StringType),
                StructField("macmd5",StringType),
                StructField("idfamd5",StringType),
                StructField("openudidmd5",StringType),
                StructField("androididmd5",StringType),
                StructField("imeisha1",StringType),
                StructField("macsha1",StringType),
                StructField("idfasha1",StringType),
                StructField("openudidsha1",StringType),
                StructField("androididsha1",StringType),
                StructField("uuidunknow",StringType),
                StructField("userid",StringType),
                StructField("iptype",IntegerType),
                StructField("initbidprice",DoubleType),
                StructField("adpayment",DoubleType),
                StructField("agentrate",DoubleType),
                StructField("lomarkrate",DoubleType),
                StructField("adxrate",DoubleType),
                StructField("title",StringType),
                StructField("keywords",StringType),
                StructField("tagid",StringType),
                StructField("callbackdate",StringType),
                StructField("channelid",StringType),
                StructField("mediatype",IntegerType)
            )
        )


        // DataFrame
        val sQLContext = new SQLContext(sc)
        sQLContext.setConf("spark.sql.parquet.compression.codec", "snappy")
        // 创建dataframe
        val dataFrame = sQLContext.createDataFrame(data, schema)


        // 保存结果
        // dataFrame.write.parquet(resultOutputPath)
        // 分区存储
        dataFrame.write.partitionBy("provincename", "cityname").parquet(resultOutputPath)

        sc.stop()
    }

}
